package com.galeno.day04;

import com.galeno.FlinkUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2014:44
 */
public class Huadong {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment locEnvWebUI = FlinkUtils.getLocEnvWebUI(configuration);
        DataStreamSource<String> lines = locEnvWebUI.socketTextStream("192.168.77.101", 8888);
        SingleOutputStreamOperator<Object> map = lines.map(Integer::parseInt);
        /**
         * 第一个参数为窗口长度,
         * 第二个时间为滑动步长
         */
        AllWindowedStream<Object, TimeWindow> windowAll = map.windowAll(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)));
        SingleOutputStreamOperator<Object> sum = windowAll.sum(0);
        sum.print();
        locEnvWebUI.execute();


    }
}
